package org.example.realtime.jtp.dws.log.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * @Title: JtpOLAPWindowDwsJob
 * @Author Lianzy
 * @Package org.example.realtime.jtp.dws.log.job
 * @Date 2025/5/25 20:19
 * @description 针对app启动日志进行当日的数据实时查询统计
 */
public class JtpOLAPWindowDwsJob {
    public static void main(String[] args) {
        //执行环境-env
        TableEnvironment tableEnv = getTableEnv();
        //映射clickhouse jtp_app_log.dwd_traffic_app_start_log 表
        createInputTable(tableEnv);

        savetoclickhouse(tableEnv);


    }

    private static void savetoclickhouse(TableEnvironment tableEnv) {
        tableEnv.executeSql(" insert into dws_traffic_app_daily_start_analysis_sink\n" +
                "                select * from start_report_table_view");

    }

    private static void createInputTable(TableEnvironment tableEnv) {
        tableEnv.executeSql("CREATE TABLE IF NOT EXISTS dwd_traffic_app_start_log_source (\n" +
                "mid STRING COMMENT '设备id',\n" +
                "area_code STRING COMMENT '地区编码',\n" +
                "brand STRING COMMENT '手机品牌',\n" +
                "channel STRING COMMENT '渠道',\n" +
                "is_new STRING COMMENT '是否首日使用，首次使用的当日，该字段值为1，过了24:00，该字段置为0',\n" +
                "model STRING COMMENT '手机型号',\n" +
                "os STRING COMMENT '操作系统',\n" +
                "uid STRING COMMENT '会员id',\n" +
                "version_code STRING COMMENT 'app版本号',\n" +
                "entry STRING COMMENT 'icon手机图标、notice 通知 、install安装后启动',\n" +
                "loading_time BIGINT COMMENT '启动加载时间',\n" +
                "open_ad_id BIGINT COMMENT ' 广告页ID',\n" +
                "open_ad_ms BIGINT COMMENT '广告总共播放时间',\n" +
                "open_ad_skip_ms BIGINT COMMENT '用户跳过广告时点',\n" +
                "ts BIGINT COMMENT '跳入时间戳'\n" +
                ")WITH(\n" +
                "'connector' = 'clickhouse',\n" +
                "'url' = 'jdbc:clickhouse://node103:8123/jtp_app_log',\n" +
                "'table' = 'dwd_traffic_app_start_log',\n" +
                "'username' = 'default',\n" +
                "'password' = '',\n" +
                "'format' = 'json')");

        /*
         *启动次数  start——count
         *启动用户数 start——user_count
         * 新用户数 nwe_user——count
         */
//        select
//                count(mid) as start_count,
//                count(distinct uid) as start_user_count,
//                count(if(is_new = '1', mid, null)) as new_user_count
//                from jtp_app_log.dwd_traffic_app_start_log_source
//        where toDate(toDateTime(ts / 1000)) = current_date()
//    group by toHour(toDateTime(ts / 1000))
        Table table = tableEnv.sqlQuery("select\n" +
                "toDate(toDateTime(ts / 1000)) as dt,\n"+
                "    count(mid) as start_count,\n" +
                "    count(distinct uid) as start_user_count,\n" +
                "    count(if(is_new = '1', mid, null)) as new_user_count\n" +
                "from dwd_traffic_app_start_log_source\n" +
                "where toDate(toDateTime(ts / 1000)) = current_date()");

        tableEnv.createTemporaryView("start_report_table_view", table);

        tableEnv.executeSql("create table dws_traffic_app_daily_start_analysis_sink\n" +
                "(\n" +
                "    -- 日期时间\n" +
                "    dt DATE COMMENT '统计日期',\n" +
                "    -- app启动信息\n" +
                "    start_count BIGINT COMMENT '启动次数',\n" +
                "    start_user_count BIGINT COMMENT '启动用户数',\n" +
                "    new_user_count BIGINT COMMENT '新增用户数'\n" +
                ")WITH(\n" +
                "'connector' = 'clickhouse',\n" +
                "'url' = 'jdbc:clickhouse://node103:8123/jtp_app_log',\n" +
                "'table' = 'dws_traffic_app_daily_start_analysis',\n" +
                "'username' = 'default',\n" +
                "'password' = '',\n" +
                "'format' = 'json')");

    }

    private static TableEnvironment getTableEnv() {
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();
        TableEnvironment tEnv = TableEnvironment.create(settings);
        //配置属性
        Configuration configuration = tEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
//        configuration.setSTRING("");
        // 返回对象
        return tEnv;
    }
}
